package com.ber.rocketmq;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;

/**
 * @Author bq8023
 * @Date 21/7/1 09:54
 * @Version 1.0
 */
@Component
public class Producer {
    private static Object s = new Object();
    private static int count = 0;

    @PostConstruct
    public void producer() throws InterruptedException, MQClientException {

        String topic = "test";
        String tag = "a";
        String group = "test_group";
        // 1024Bytes
        String text = "这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字最后";
        /*
         * Instantiate with a producer group name.
         */
        DefaultMQProducer producer = new DefaultMQProducer(group);

        // 设置NameServer地址
        producer.setNamesrvAddr("58.59.43.108:9876");

        /*
         * Launch the instance.
         */

        producer.start();
        for (int i = 0; i < 1; i++) {
            new Thread() {
                @Override
                public void run() {
                    synchronized(s){
                        count += 1;
                        System.err.println("New thread #"+count);
                    }

                    while (true) {
                        Message msg = null;
                        try {
                            msg = new Message(topic, tag, text.getBytes(RemotingHelper.DEFAULT_CHARSET));
                            SendResult sendResult = producer.send(msg);
                            System.out.printf("%s%n", sendResult);
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        } catch (MQBrokerException e) {
                            e.printStackTrace();
                        } catch (RemotingException e) {
                            e.printStackTrace();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (MQClientException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }.start();
        }

        while (true) {
            Thread.sleep(60000);
        }
    }

    public static void main(String[] args) throws Exception {

        String topic = "test";
        String tag = "a";
        String group = "test_group";
        // 1024Bytes
        String text = "这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字这是五个字最后";
        /*
         * Instantiate with a producer group name.
         */
        DefaultMQProducer producer = new DefaultMQProducer(group);

        // 设置NameServer地址
        producer.setNamesrvAddr("58.59.43.108:9876");

        /*
         * Launch the instance.
         */

        producer.start();
        for (int i = 0; i < 1; i++) {
            new Thread() {
                @Override
                public void run() {
                    synchronized(s){
                        count += 1;
                        System.err.println("New thread #"+count);
                    }

                    while (true) {
                        Message msg = null;
                        try {
                            msg = new Message(topic, tag, text.getBytes(RemotingHelper.DEFAULT_CHARSET));
                            SendResult sendResult = producer.send(msg);
                            System.out.printf("%s%n", sendResult);
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        } catch (MQBrokerException e) {
                            e.printStackTrace();
                        } catch (RemotingException e) {
                            e.printStackTrace();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (MQClientException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }.start();
        }

        while (true) {
            Thread.sleep(60000);
        }
    }
}
